package cn.dmp.charts.arealDistribution

import java.util.Properties

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.{DataFrame, SQLContext}

object ArealDistribution171106V1 {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setAppName("DataDistributionProvinceAndCity").setMaster("local[3]")
    val sc: SparkContext = new SparkContext(conf)
    //包装sparkcontext后增强
    val sqlContext= new SQLContext(sc)

    val logDf: DataFrame = sqlContext.read.load(args(0))
//选出字段
    val select: DataFrame = logDf.select("provincename","cityname","requestmode","processnode","iseffective","isbilling","isbid","iswin","adorderid","winprice","adpayment")
//聚合
    select.registerTempTable("t_distribution")
    val sql = "select provincename,cityname," +
      " sum(case when (requestmode =1 and processnode>=1) then 1 else 0 end) primaryRequest," +
      " sum(case when (requestmode =1 and processnode>=2) then 1 else 0 end ) effectiveRequest," +
      " sum(case when (requestmode =1 and processnode>=3) then 1 else 0 end ) AdRequest, " +
      " sum(case when (iseffective = 1 and isbilling = 1 and isbid = 1 and adorderid !=0) then 1 else 0 end) biddingTimes," +
      " sum(case when (iseffective = 1 and isbilling = 1 and iswin = 1 ) then 1 else 0 end) succesedBiddingTimes, " +
      " sum(case when (requestmode =2 and iseffective =1 ) then 1 else 0 end ) showTimes, " +
      " sum(case when (requestmode =3 and iseffective =1 ) then 1 else 0 end ) clickTimes, " +
      " sum(if((iseffective=1 and isbilling=1 and iswin =1),winprice,0))/1000  AdConsume, " +
      " sum(if((iseffective=1 and isbilling=1 and iswin =1),adpayment,0))/1000  AdCost " +
      "from t_distribution" +
      " group by provincename,cityname "
        val props = new Properties()
        props.put("user","root")
        props.put("password","123456")
        sqlContext.sql(sql).write.mode("append").jdbc("jdbc:mysql://localhost:3306/npm?useUnicode=true&characterEncoding=utf-8", "ArealDistribution171106V1", props)

//  sqlContext.sql("select provincename,cityname,count(*) " +
//    "from t_distribution" +
//    " where requestmode =1 and processnode >= 1 " +
//    "group by provincename,cityname").show()


        sc.stop()

  }
}
